#include "StdAfx.h"
#include "TdFastExport.h"

#include "TdUtils.h"
#include "connection.h"
#include "schema.h"
#include "DMLGroup.h"

using namespace System::Collections::Generic;
using namespace System::Net;
using namespace System::Net::Sockets;
using namespace System::Text;
using namespace Teradata::Client::Provider;
using namespace teradata::client::API;
namespace TPT = TdParallelTransportApi;

TPT::TdLogLevel
TPT::TdFastExport::LogLevel::get()
{
	return m_LogLevel;
}

void
TPT::TdFastExport::LogLevel::set(TPT::TdLogLevel value)
{
	m_LogLevel = value;
}

TPT::TdFastExport::TdFastExport(TdBulkConnection^ connection) :
	m_Connection(connection)
{
}

System::String^
TPT::TdFastExport::SourceTableName::get()
{
	return this->m_SourceTableName;
}

void
TPT::TdFastExport::SourceTableName::set(System::String^ value)
{
	this->m_SourceTableName = value;
}

System::Boolean CheckBit(array<unsigned char>^ data, int index)
{
	int offset = (index / 8);
	int bit = 7 - (index % 8);

	return ((data[offset] >> bit) & 1) == 1;
}

array<unsigned char>^ MarshalArray(char* ptr, int len)
{
	array<unsigned char>^ ary = gcnew array<unsigned char>(len);

	for (int i = 0; i < len; i++)
	{
		ary[i] = (unsigned char)ptr[i];
	}

	return ary;
}

void FECheckStatus(teradata::client::API::Connection *conn, TD_StatusCode status)
{
	if (status >= TD_Error)
	{
		char* msg;
		TD_ErrorType type;
		conn->GetErrorInfo(&msg, &type);
		throw gcnew System::InvalidOperationException("Teradata Error (" + ((int)type).ToString() + "): " + gcnew System::String(msg));
	}
}

void
TPT::TdFastExport::Initialize(array<TdColumn^>^ sourceColumns, int restartmode, void** conn_ptr, void** schema_ptr)
{
	this->username = (char*)(void*)Marshal::StringToHGlobalAnsi(this->m_Connection->Username);
	this->password = (char*)(void*)Marshal::SecureStringToGlobalAllocAnsi(this->m_Connection->Password);

	array<IPAddress^>^ entry = Dns::GetHostAddresses(this->m_Connection->Server);
	System::String^ serverIP;
	for (int i = 0; i < entry->Length; i++)
	{
		if (entry[i]->AddressFamily == Sockets::AddressFamily::InterNetwork)
		{
			serverIP = entry[i]->ToString();
			break;
		}
	}

	if (System::String::IsNullOrEmpty(serverIP))
	{
		throw gcnew System::InvalidOperationException("The Server Name Was Not Found.");
	}

	this->server = (char*)(void*)Marshal::StringToHGlobalAnsi(serverIP);
	this->database = (char*)(void*)Marshal::StringToHGlobalAnsi(this->m_Connection->Database);
	this->target = (char*)(void*)Marshal::StringToHGlobalAnsi(this->m_SourceTableName);

	Connection *conn = new Connection(TD_UTF8_ENCODING);

	/**********************************************
	* Add Attributes
	**********************************************/
	conn->AddAttribute(TD_SYSTEM_OPERATOR, TD_EXPORT);

	conn->AddAttribute(TD_USER_NAME, this->username);
	conn->AddAttribute(TD_USER_PASSWORD, this->password);
	conn->AddAttribute(TD_TDP_ID, this->server);
	conn->AddAttribute(TD_WORKINGDATABASE, this->database);
	conn->AddAttribute(TD_TARGET_TABLE, this->target);
	conn->AddAttribute(TD_ROBUST, "No");
	conn->AddAttribute(TD_CHARSET, "UTF16");
	conn->AddAttribute(TD_MIN_SESSIONS, 1);
	conn->AddAttribute(TD_MAX_SESSIONS, 4);

	if (restartmode == 1)
	{
		conn->AddAttribute(TD_RESTARTMODE, 1);
	}

	this->logLevel = NULL;

	if (this->m_LogLevel != TdLogLevel::Off)
	{
		this->logLevel = GetLogLevelString(this->m_LogLevel);
		conn->AddAttribute(TD_NOTIFY_LEVEL, this->logLevel);
		conn->AddAttribute(TD_NOTIFY_METHOD, "Msg");
	}

	/**********************************************
	* Add Schema
	**********************************************/
	System::Collections::IEnumerator^ colEnum = sourceColumns->GetEnumerator();

	bool first = true;

	StringBuilder^ selectSB = gcnew StringBuilder();
	Schema *schema = new Schema("output");
	
	selectSB->Append("SELECT ");

	while (colEnum->MoveNext())
	{
		if (!first)
		{
			selectSB->Append(", ");
		}
		else
		{
			first = false;
		}
		
		TdColumn^ current = (TdColumn^)colEnum->Current;

		char* fieldName = (char*)(void*)Marshal::StringToHGlobalAnsi(current->ColumnName);
		schema->AddColumn(fieldName, TdColumn::GetTdDataType(current->DataType), current->Size, current->Precision, current->Scale);
		selectSB->Append(current->ColumnName);
		Marshal::FreeHGlobal((System::IntPtr)fieldName);
	}

	selectSB->Append(" FROM ");
	selectSB->Append(this->SourceTableName);
	selectSB->Append(";");

	this->select = (char*)(void*)Marshal::StringToHGlobalAnsi(selectSB->ToString());
	conn->AddAttribute(TD_SELECT_STMT, select);

	conn->AddSchema(schema);

	FECheckStatus(conn, conn->Initiate());

	*conn_ptr = (void*)conn;
	*schema_ptr = (void*)schema;
}

void
TPT::TdFastExport::ReadFromServer(
		array<TdColumn^>^ sourceColumns,
		System::Action<array<System::Object^>^>^ rowReceived)
{
	Connection* conn;
	Schema* schema;

	this->Initialize(sourceColumns, 0, (void**)&conn, (void**)&schema);

	try
	{
		while (true)
		{
			char* ptr = NULL;
			TD_Length len = 0;

			TD_StatusCode status = conn->GetRow(&ptr, &len);

			if (status == TD_END_Method)
			{
				break;
			}
			else
			{
				FECheckStatus(conn, status);
			}

			array<unsigned char>^ buffer = MarshalArray(ptr, len);
			array<System::Object^>^ values = gcnew array<System::Object^>(sourceColumns->Length);

			int offset = (sourceColumns->Length + 7) / 8;
			for (int i = 0; i < sourceColumns->Length; i++)
			{
				if (CheckBit(buffer, i))
				{
					if (sourceColumns[i]->DataType == TdType::VarByte || sourceColumns[i]->DataType == TdType::VarChar)
					{
						offset += 2;
					}
					else
					{
						offset += sourceColumns[i]->Size;
					}
				}
				else
				{
					switch (sourceColumns[i]->DataType)
					{
					case TdType::Byte:
						if (sourceColumns[i]->Length > 1)
						{
							array<System::Byte>^ ary = gcnew array<System::Byte>(sourceColumns[i]->Length);

							for (int j = 0; j < sourceColumns[i]->Length; j++)
							{
								ary[j] = (System::Byte)buffer[offset + j];
							}

							values[i] = ary;
						}
						else
						{
							values[i] = (System::Byte)buffer[offset];
						}

						offset += sourceColumns[i]->Length;

						break;
					case TdType::VarByte:
						{
							int varlen = (int)System::BitConverter::ToUInt16(buffer, offset);
							offset += 2;
							array<System::Byte>^ ary = gcnew array<System::Byte>(varlen);
							for (int j = 0; j < varlen; j++)
							{
								ary[j] = (System::Byte)buffer[offset + j];
							}

							values[i] = ary;
							offset += varlen;
						}

						break;
					case TdType::Char:
						values[i] = Encoding::Unicode->GetString(buffer, offset, sourceColumns[i]->Size);

						offset += sourceColumns[i]->Size;

						break;
					case TdType::VarChar:
						{
							int varlen = (int)System::BitConverter::ToUInt16(buffer, offset);
							offset += 2;

							values[i] = Encoding::Unicode->GetString(buffer, offset, varlen);
							offset += varlen;
						}

						break;
					case TdType::ByteInt:
						values[i] = (System::SByte)buffer[offset];
						offset++;

						break;
					case TdType::SmallInt:
						values[i] = System::BitConverter::ToInt16(buffer, offset);
						offset += 2;

						break;
					case TdType::Integer:
						values[i] = System::BitConverter::ToInt32(buffer, offset);
						offset += 4;

						break;
					case TdType::BigInt:
						values[i] = System::BitConverter::ToInt64(buffer, offset);
						offset += 8;

						break;
					case TdType::Decimal:
						{
							int size = sourceColumns[i]->Size;
							array<System::Byte>^ decBytes = gcnew array<System::Byte>(size);
							System::Boolean negative = false;

							// Check the sign of the most signficant byte
							if ((buffer[offset + size - 1] & 128) != 128)
							{
								// Positive
								System::Array::Copy(buffer, offset, decBytes, 0, size);
							}
							else
							{
								negative = true;

								// Negative
								for (int j = 0; j < size; j++)
								{
									decBytes[j] = ~((System::Byte)buffer[offset + j]);
								}

								int carry = 1;
								for (int j = 0; j < size && carry == 1; j++)
								{
									int add = ((int)decBytes[j]) + carry;

									if (add < 256)
									{
										decBytes[j] = (System::Byte)add;
										carry = 0;
									}
									else
									{
										decBytes[j] = 0;
										carry = 1;
									}
								}
							}

							for (int j = 12; j < size && j < 16; j++)
							{
								if (decBytes[j] > 0)
								{
									throw gcnew System::OverflowException("This Decimal is Too Large to Represent as a .Net Decimal");
								}
							}

							int low = 0, mid = 0, high = 0;

							if (size == 1)
							{
								low = decBytes[0];
							}
							else if (size == 2)
							{
								low = System::BitConverter::ToUInt16(decBytes, 0);
							}
							else if (size == 4)
							{
								low = System::BitConverter::ToInt32(decBytes, 0);
							}
							else if (size == 8)
							{
								low = System::BitConverter::ToInt32(decBytes, 0);
								mid = System::BitConverter::ToInt32(decBytes, 4);
							}
							else if (size == 16)
							{
								low = System::BitConverter::ToInt32(decBytes, 0);
								mid = System::BitConverter::ToInt32(decBytes, 4);
								high = System::BitConverter::ToInt32(decBytes, 8);
							}

							values[i] = gcnew System::Decimal(low, mid, high, negative, (System::Byte)sourceColumns[i]->Scale);
							offset += size;
						}

						break;
					case TdType::Double:
						values[i] = System::BitConverter::ToDouble(buffer, offset);
						offset += 8;

						break;
					case TdType::Date:
						{
							int dateInt = System::BitConverter::ToInt32(buffer, offset);

							int year = (int)System::Math::Floor((System::Double)dateInt / 10000.0) + 1900;
							int month = (dateInt - ((year - 1900) * 10000)) / 100;
							int day = dateInt - (month * 100) - ((year - 1900) * 10000);

							values[i] = gcnew System::DateTime(year, month, day);

							offset += 4;
						}

						break;
					}
				}
			}

			rowReceived(values);
		}
	}
	finally
	{
		try
		{
			conn->Terminate();
		}
		catch (System::Exception^)
		{
			// Squish
		}

		delete conn;
		delete schema;

		this->CleanUp();
	}
}

void
TPT::TdFastExport::CleanUp()
{
		Marshal::FreeHGlobal((System::IntPtr)this->username);
		Marshal::FreeHGlobal((System::IntPtr)this->password);
		
		Marshal::FreeHGlobal((System::IntPtr)this->server);
		Marshal::FreeHGlobal((System::IntPtr)this->database);
		Marshal::FreeHGlobal((System::IntPtr)this->target);

		if (logLevel != NULL)
		{
			Marshal::FreeHGlobal((System::IntPtr)this->logLevel);
		}

		Marshal::FreeHGlobal((System::IntPtr)this->select);
}
